package com.mjf.spark.day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子-groupBy
 *    按照传入函数的返回值，对RDD中的元素进行分组
 */
object Spark05_Transformation_groupBy {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark05_Transformation_groupBy")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 2)

    println("===============执行groupBy分组前===============")
    rdd.mapPartitionsWithIndex(
      (index, datas) => {
        println(index + "--->" + datas.mkString(","))
        datas
    }).collect()

    val newRDD: RDD[(Int, Iterable[Int])] = rdd.groupBy(_ % 3)

    println("===============执行groupBy分组前===============")
    newRDD.mapPartitionsWithIndex(
      (index, datas) => {
        println(index + "--->" + datas.mkString(","))
        datas
      }).collect()

    // 关闭连接
    sc. stop()

  }
}
